package com.bckj.fastboot.mq.ons;

import com.aliyun.openservices.ons.api.*;
import com.bckj.fastboot.core.util.TransactionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractOnsProducer<T> {
    private static final Logger log = LoggerFactory.getLogger(AbstractOnsProducer.class);

    public AbstractOnsProducer() {
    }

    /**
     * 发送订单消息（支持同步/异步）
     * @param msg 消息体
     * @param sendCallback 可选回调（为null时同步发送）
     */
    public void send(T msg, SendCallback sendCallback) {
        try {
            Message message = buildMessage(msg);
            if (sendCallback != null) {
                getProducer().sendAsync(message, wrapCallback(msg, sendCallback));
            } else {
                getProducer().send(message); // 同步发送
            }
        } catch (Exception e) {
            log.error("[MQ] 订单消息发送失败 | msg:{}", msg, e);
            throw new MqSendException("订单消息发送失败", e);
        }
    }

    protected abstract Message buildMessage(T msg);

    protected abstract Producer getProducer();

    /**
     * 事务提交后发送消息
     * @param msg 消息参数
     */
    public void afterSend(T msg, SendCallback sendCallback) {
        TransactionUtils.afterCommit(() -> send(msg, sendCallback));
    }

    protected SendCallback wrapCallback(T msg, SendCallback original) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                AbstractOnsProducer.log.info("[" + msg.getClass() + "] 成功,messageId:{}", sendResult.getMessageId());
                original.onSuccess(sendResult);
            }
            @Override
            public void onException(OnExceptionContext context) {
                AbstractOnsProducer.log.error("[" + msg.getClass() + "] 失败,messageId:{}, ErrorMsg:{}, params:{}",
                        context.getMessageId(), context.getException().getMessage(), msg);
                original.onException(context);
                // 可在此添加重试逻辑
            }
        };
    }
}
